package com.shanks.framework.mq.rocketmq.config;

import com.shanks.framework.mq.configuration.BaseMQConfiguration;
import com.shanks.framework.mq.enums.MQZone;
import com.shanks.framework.mq.exception.MQException;
import com.shanks.framework.mq.rocketmq.consumer.RocketMQConsumerAttribute;
import com.shanks.framework.mq.rocketmq.producer.RocketMQProducerImpl;
import jakarta.annotation.Resource;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;

@ComponentScan("com.shanks.framework.mq")
public class RocketMQConfiguration extends BaseMQConfiguration {

    @Resource
    private RocketMQProducerInitializer rocketMQProducerInitializer;

    /**
     * 注入默认的ack处理器，子类可重写
     *
     * @return
     */
    @Bean
    protected RocketMQConsumerInitializer rocketMQConsumerInitializer() {
        return new RocketMQConsumerInitializer(ackMQProcessor());
    }

    /**
     * 从配置读取生产者配置并启动生产者
     *
     * @param propertyPrefix
     * @param mqZone
     * @return
     * @throws MQException
     */
    protected RocketMQProducerImpl createAndRegisterMQProducer(String propertyPrefix, MQZone mqZone) throws MQException {
        return rocketMQProducerInitializer.createAndResiterMQProducer(propertyPrefix, mqZone);
    }


    protected DefaultMQPushConsumer createAndStartMQConsumer(RocketMQConsumerAttribute rocketMQConsumerAttribute) throws MQException {
        return rocketMQConsumerInitializer().createAndStartMQConsumer(rocketMQConsumerAttribute);
    }


    protected RocketMQConsumerAttribute newConsumerAttribute(String propertyPrefix, MQZone mqZone) throws MQException {
        return rocketMQConsumerInitializer().buildRocketMQConsumerAttribute(propertyPrefix, mqZone);
    }
}
